package rdd.operate;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Serializable;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.List;

//sortByKey也是在元组，也就是kv类型下操作的
//通过key直接进行排序
//但是sotyByKey需要key是能排序的，如果是对象或者某些不能直接比较的，需要实现comparable接口，重写ComparaTo方法
public class Spark59_Operate_sortByKey {
    public static void main(String[] args) {
        final SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("spark");
        final JavaSparkContext jsc = new JavaSparkContext(conf);

        final List<Tuple2<User,Integer>> datas = new ArrayList<>();

        User user1 = new User();
        user1.age = 30;
        user1.amount = 2000;

        User user2 = new User();
        user2.age = 40;
        user2.amount = 3000;

        User user3 = new User();
        user3.age = 30;
        user3.amount = 3000;

        User user4 = new User();
        user4.age = 40;
        user4.amount = 2500;

        datas.add(new Tuple2<User,Integer>(user1,1));
        datas.add(new Tuple2<User,Integer>(user2,3));
        datas.add(new Tuple2<User,Integer>(user3,4));
        datas.add(new Tuple2<User,Integer>(user4,2));

        final JavaRDD<Tuple2<User,Integer>> rdd = jsc.parallelize(datas);
        final JavaPairRDD<User,Integer> pairRDD = rdd.mapToPair(t -> t);
        pairRDD.sortByKey().collect().forEach(System.out::println);
    }
}

class User implements Serializable,Comparable<User> {
    int age = 0;
    int amount = 0;

    @Override
    public String toString() {
        return "User{" +
                "age=" + age +
                ", amount=" + amount +
                '}';
    }

    @Override
    //返回<0,代表当前对象比比较的对象小
    //返回>0,代表当前对象比比较的对象大
    //返回==0,代表当前对象于比较对象一样大
    public int compareTo(User other) {
        //只按照年龄排序，因为返回的是int，可以直接将两个age进行做差
        //return this.age - other.age
        //先年龄升序，若年龄相等，就amount升序
        if(this.age < other.age ){
            return -1;
        }else if(this.age > other.age){
            return 1;
        }else{
            if(this.amount < other.amount){
                return -1;
            }else if(this.amount > other.amount){
                return 1;
            }else{
                return 0;
            }
        }
    }
}
